package org.niit.stream

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.collection.mutable

//需求：循环创建几个RDD ,将RDD放入到队列当中。通过 SparkStream 创建DStream 计算wordCount
object SparkStreaming_02 {
  def main(args: Array[String]): Unit = {
   //环境准备
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("streaming")
    val ssc = new StreamingContext(sparkConf, Seconds(3))
    ssc.sparkContext.setLogLevel("ERROR")

    //创建RDD队列 可变队列 队列 ：先进先出
    val rddQueue =  new mutable.Queue[RDD[Int]]()
    //创建队列输入流
    val inputStream: InputDStream[Int] = ssc.queueStream(rddQueue, oneAtATime = false)
    // 套接字输入流：用来监听端口数据的变化  val lines: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)

    //分析数据
    val mapStream: DStream[(Int, Int)] = inputStream.map((_, 1))
    val reduceStream: DStream[(Int, Int)] = mapStream.reduceByKey(_ + _)

    //打印结果
    reduceStream.print(300)
    //环境准备
    ssc.start()
    //循环创建 RDD
    for (i <- 1 to 50){
      rddQueue += ssc.sparkContext.makeRDD(1 to 100,10)
      Thread.sleep(2000)
    }

    ssc.awaitTermination()//阻塞的作用 目的接收数据
  }
}
